/**
 *
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.openness.crawler.zookeeper;

import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A base class for protocol implementations which provides a number of higher
 * level helper methods for working with ZooKeeper along with retrying
 * synchronous operations if the connection to ZooKeeper closes such as
 * {@link #retryOperation(ZooKeeperOperation)}
 * 
 */
public class ProtocolSupport {

	private static final Logger LOGGER = LoggerFactory
			.getLogger(ProtocolSupport.class);

	private long retryDelay = 500L;

	private int retryCount = 10;

	/**
	 * get the retry delay in milliseconds
	 * 
	 * @return the retry delay
	 */
	public long getRetryDelay() {
		return retryDelay;
	}

	/**
	 * Sets the time waited between retry delays
	 * 
	 * @param retryDelay
	 *            the retry delay
	 */
	public void setRetryDelay(long retryDelay) {
		this.retryDelay = retryDelay;
	}

	/**
	 * Allow derived classes to perform some custom closing operations to
	 * release resources
	 */
	protected void doClose() {
	}

	/**
	 * Perform the given operation, retrying if the connection fails
	 * 
	 * @return object. it needs to be cast to the callee's expected return type.
	 */
	protected <T> T retryOperation(ZooKeeperOperation<T> operation)
			throws Exception {
		KeeperException exception = null;

		for (int i = 0; i < retryCount; i++) {
			try {
				return operation.execute();
			} catch (KeeperException.SessionExpiredException e) {
				LOGGER.warn("Session expired: " + e.toString());

				throw e;
			} catch (KeeperException.ConnectionLossException e) {
				LOGGER.warn("Connection loss: " + e.toString());

				if (exception == null) {
					exception = e;
				}

				LOGGER.debug("Attempt " + i
						+ " failed with connection loss so "
						+ "attempting to reconnect");

				retryDelay(i);
			}
		}

		throw exception;
	}

	/**
	 * Performs a retry delay if this is not the first attempt
	 * 
	 * @param attemptCount
	 *            the number of the attempts performed so far
	 */
	private void retryDelay(int attemptCount) {
		if (attemptCount > 0) {
			try {
				Thread.sleep(attemptCount * retryDelay);
			} catch (InterruptedException e) {
				LOGGER.debug("Failed to sleep: " + e, e);
			}
		}
	}

}
